package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

// 虚拟主机，管理自己得交换机，队列，绑定，消息，数据，同时提供api给上层调用
// 针对 VirtualHost 这个类, 作为业务逻辑的整合者, 就需要对于代码中抛出的异常进行处理了
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();
    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public VirtualHost(String name) {
        this.virtualHostName = name;

        // 对于 MemoryDataCenter 来说, 不需要额外的初始化操作的. 只要对象 new 出来就行了
        // 但是, 针对 DiskDataCenter 来说, 则需要进行初始化操作. 建库建表和初始数据的设定
        diskDataCenter.init();
        // 另外还需要针对硬盘的数据, 进行恢复到内存中

        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | ClassNotFoundException | MqException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败!");
        }
    }

    // 创建交换机
    // 不存在创建，存在返回
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable,
                                   boolean autoDelete, Map<String, Object> arguments) {
        // 交换机得名字，加上虚拟主机作为前缀
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker) {
                // 1.判断交换机是否存在，通过内存查询
                Exchange existExchange = memoryDataCenter.getExchange(exchangeName);
                if (existExchange != null) {
                    System.out.println("[VirtualHost] 交换机已经存在! exchangeName=" + exchangeName);
                    return true;
                }
                // 2.不存在，创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                // 3.保存到硬盘
                if(durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                // 4.保存到内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建成功! exchangeName:" + exchangeName);
            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 交换机创建失败! exchangeName:" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除交换机
    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1.判断交换机是否存在
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if(toDelete == null) {
                    throw new MqException("交换机不存在无法删除!");
                }
                // 2.删除硬盘数据
                if(toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                // 3.删除内存数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功! exchangeName:" + exchangeName);
            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName:" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 创建队列
    public boolean queueDeclare(String queueName,boolean exclusive,boolean durable,
                                boolean autoDelete,Map<String,Object> arguments) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1.判断队列是否存在
                MSGQueue existQueue = memoryDataCenter.getQueue(queueName);
                if(existQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在! queueName:" + queueName);
                    return false;
                }
                // 2.创建队列
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 3.写硬盘
                if(durable) {
                    diskDataCenter.insertQueue(queue);
                }
                // 4.写内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName:" + queueName);
            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName:" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除队列
    public boolean deleteQueue(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1.检查队列是否存在
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null) {
                    throw new MqException("队列不存在! queueName:" + queueName);
                }
                // 2.删除硬盘数据
                if(queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                // 3.删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功! queueName:" + queueName);
            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败! queueName:" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueBind(String queueName,String exchangeName,String routingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1.检测绑定是否存在
                    Binding existBinding = memoryDataCenter.getBinding(exchangeName,queueName);
                    if(existBinding != null) {
                        throw new MqException("[VirtualHost] 绑定关系已存在! queueName:" + queueName + ",exchangeName:" + exchangeName);
                    }
                    // 2.验证Binding是否合法
                    if(!router.checkBindingKey(routingKey)) {
                        throw new MqException("[VirtualHost] routingKey不合法! routingKey:" + routingKey);
                    }
                    // 3.创建Binding
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(routingKey);
                    // 4.获取交换机和队列
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null) {
                        throw new MqException("[VirtualHost] 队列不存在! queueName:" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在! exchangeName:" + exchangeName);
                    }
                    // 5.硬盘
                    if(queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    // 6.内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功！exchangeName:" + exchangeName + " queueName:" + queueName);
                }
            }
            return true;
        }catch(Exception e) {
            System.out.println("[VirtualHost] 队列绑定失败 exchangeName:" + exchangeName + " queueName:" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除绑定
    public boolean queueUnbind(String exchangeName, String queueName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1.检测绑定是否存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(binding == null) {
                        throw new MqException("[VirtualHost] 绑定不存在! exchangeName:" + exchangeName + " queueName:" + queueName);
                    }
                    // 2.无论绑定是否持久化，都尝试从内存删下，就算不存在，也是没有副作用
                    diskDataCenter.deleteBinding(binding);
                    // 3.删除内存绑定
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 队列绑定删除成功");
                }
            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 队列绑定删除失败 exchangeName:" + exchangeName + " queueName:" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 发送消息到指定的交换机/队列中
    public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body) {
        try {
            // 1.转换名字
            exchangeName = virtualHostName + exchangeName;
            // 2.检测routingKey是否合法
            if(!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey不合法! routingKey:" + routingKey);
            }
            // 3.检测交换机是否存在
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在! exchangeName:" + exchangeName);
            }
            // 4.判断交换机类型
            if(exchange.getType() == ExchangeType.DIRECT) {
                // 4.1.直接交换机
                // routingKey和队列名相同，此时可无视绑定关系
                String queueName = virtualHostName + routingKey;
                // 5.构造消息
                Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                // 6.查找队列名对象的
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在! queueName:" + queueName);
                }
                // 7.队列存在，队列中写入
                sendMessage(queue,message);
            }else {
                // 4.2.fanout 和 topic交换机
                // 5.找到该交换机关联的所有绑定，遍历这些对象
                ConcurrentHashMap<String, Binding> bindingMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String,Binding> entry: bindingMap.entrySet()) {
                    // 获取绑定对象，判断队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        // 不抛出异常，希望不要因为一个队列失败，影响其他队列消息的传输
                        System.out.println("[VirtualHost] basicPublish发送消息时，队列不存在! queueName:" + binding.getQueueName());
                        continue;
                    }
                    // 构造消息对象
                    Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                    // 判断消息是否能转发给队列
                    //  fanout交换机，所有队列转发
                    //  topic交换机，根据routingKey判断是否转发
                    if(!router.route(exchange.getType(),binding,message)) {
                        continue;
                    }
                    // 真正转发
                    sendMessage(queue,message);
                }
            }
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 发送消息失败 exchangeName:" + exchangeName + " routingKey:" + routingKey);
            e.printStackTrace();;
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        // 发送消息，就是把消息写道硬盘和内存上
        int deliverMode = message.getDeliveryMode();
        // 1.不持久 2.持久
        if(deliverMode == 2) {
            diskDataCenter.sendMessage(queue,message);
        }
        memoryDataCenter.sendMessage(queue,message);

        //  通知消费者可消费了
        consumerManager.notifyConsumer(queue.getName());
    }

    // 订阅消息
    // 添加一个队列的订阅者，队列收到消息后，把消息推送到对应的订阅者
    public boolean basicConsume(String consumerTag,String queueName, boolean autoAck, Consumer consumer) {
        // 构造一个 ConsumerEnv 对象，把这个对应的队列找到，再把这个 Consumer 对象添加到队列中
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] 添加消费者成功! queueName:" + queueName);
            return true;
        }catch (Exception e) {
            System.out.println("[VirtualHost] 添加消费者失败! queueName:" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean basicAck(String queueName,String messageId) {
        queueName = virtualHostName + queueName;
        try {
            // 获取消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if(message == null) {
                throw new MqException("[virtualHost] basicAck 消息不存在! messageId:" + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null) {
                throw new MqException("[virtualHost] basicAck 队列不存在! queueName:" + queueName);
            }
            // 硬盘
            if(message.getDeliveryMode() == 2) {
                diskDataCenter.deleteMessage(queue,message);
            }
            // 消息中心
            memoryDataCenter.removeMessage(messageId);
            // 待确认集合中数据
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            System.out.println("[virtualHost] basicAck 消息确认成功! queueName " + queueName +
                    "messageId:" + messageId);
            return true;
        }catch (Exception e) {
            System.out.println("[virtualHost] basicAck 消息确认失败! queueName " + queueName +
                    "messageId:" + messageId);
            e.printStackTrace();
            return false;
        }
    }
}
